package com.yingjia.yingjiauniversity.rxbus;


import com.my.common.utils.GenericReflectUtil;

import io.reactivex.Observable;
import io.reactivex.functions.Predicate;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.Subject;

/**
 * <RxBus>
 *
 * @author: guoxiangxun
 * @date: 17-7-6 下午4:10
 * @version: v1.0
 */
public class RxBus {
    //非背压处理
    private final Subject<Object> bus;
    private static volatile RxBus INSTANCE;

    private RxBus() {
        //非背压处理
        bus = PublishSubject.create().toSerialized();
    }

    public static RxBus get() {
        if (null == INSTANCE) {
            synchronized (RxBus.class) {
                if (null == INSTANCE) {
                    INSTANCE = new RxBus();
                }
            }
        }
        return INSTANCE;
    }

    public void post(Object o) {
        bus.onNext(o);
    }

    public boolean hasObservable() {
        return bus.hasObservers();
    }

    /*
     * 转换为特定类型的Obserbale
     */
    public <T> Observable<T> toObservable(Class<T> clazz) {
        return bus.ofType(clazz);
    }


    /**
     * 支持泛型
     *
     * @param event
     * @param <T>
     * @return
     */
    public <T extends Event> Observable<T> toObservable(final T event) {
        Class<T> clazz = GenericReflectUtil.getClazz(event.getType());
        return bus.ofType(clazz)
                .filter(new Predicate<T>() {
                    @Override
                    public boolean test(T tEvent) throws Exception {
                        return tEvent.getTopic().equals(event.getTopic());
                    }
                });
    }
}
